#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <sys/types.h>
#include "hdfs.h"

struct DCmonitor{
	int nr_hosts, nr_files, rep;
	char **host;
	char **file;
	int **map;
};

/*
int DCscheduler_getHostID(struct DCmonitor *p, char *host)
{
	assert(p);
	assert(host);
	int i;
	for(i=0; i<p->nr_hosts; i++){
		if(!strcmp(p->host[i], host)){
			return i;
		}
	}
	assert(0);
	return -1;
}
*/

int DCscheduler_getHostID(struct DCmonitor *p, char *host)
{
	assert(p);
	assert(host);
	int i, x;
	for (i = 0; i<p->nr_hosts; i++){
		if (!strncmp(host, p->host[i], strlen(host))){
			return i;
		}
	}
	return -1;
}

struct DCmonitor *DCmonitor_create(char *host[], char *hostFile, char *file, int nhosts, int nfiles, int rep)
{
	assert(host);
	assert(file);

	struct DCmonitor *monitor = (struct DCmonitor *)malloc(sizeof(*monitor));
	assert(monitor);
	memset(monitor, 0, sizeof(*monitor));

	int i, len, index;

	FILE * fp;
	char * line = NULL;
	size_t lenn = 0;
	ssize_t read;

	monitor->nr_hosts = nhosts;
	monitor->nr_files = nfiles;
	monitor->rep = rep;

	/* alloc memory for host */
	monitor->host = (char **)malloc(sizeof(char *)* (monitor->nr_hosts + 1));
	assert(monitor->host);
	memset(monitor->host, 0, sizeof(char *)* (monitor->nr_hosts + 1));


	fp = fopen(hostFile, "r");
	if (fp == NULL)
		exit(EXIT_FAILURE);

	index = 0;
	while ((read = getline(&line, &lenn, fp)) != -1) {
		if (index < nhosts) {
			monitor->host[index] = (char *)malloc(sizeof(char)* (read - 1));
			assert(monitor->host[index]);
			memset(monitor->host[index], 0, sizeof(char)* (read - 1));
			strcpy(monitor->host[index], line); index++;
		}

	}

	/*	for(int i =0; i< index;i++)
	printf("%s, %d", monitor->host[i],strlen(monitor->host[i]));
	printf(" monitor->host[i],strlen(monitor->host[i])");
	*/

	/* alloc memory for file */
	monitor->file = (char **)malloc(sizeof(char *)* (monitor->nr_files + 1));
	assert(monitor->file);
	memset(monitor->file, 0, sizeof(char *)* (monitor->nr_files + 1));
	for (i = 0; i<monitor->nr_files; i++){
		//len = strlen(file[i]);

		char fileName[80];
		len = sprintf(fileName, "%s%d\n", file, i);

		monitor->file[i] = (char *)malloc(sizeof(char)* (len + 1));
		assert(monitor->file[i]);
		memset(monitor->file[i], 0, sizeof(char)* (len + 1));
		strcpy(monitor->file[i], fileName);
	}

	/* alloc memory for map (file --> hosts)*/
	monitor->map = (int **)malloc(sizeof(int *)* (monitor->nr_files + 1));
	assert(monitor->map);
	memset(monitor->map, 0, sizeof(int *)* (monitor->nr_files + 1));
	for (i = 0; i<monitor->nr_files; i++){
		//len = monitor->nr_files;
		monitor->map[i] = (int *)malloc(sizeof(int)* (monitor->rep));
		assert(monitor->map[i]);
		memset(monitor->map[i], 0, sizeof(int)* (monitor->rep));
	}

	/* fill locations */
	hdfsFS fs = hdfsConnect("default", 0);
	if (!fs) {
		fprintf(stderr, "Oops! Failed to connect to hdfs!\n");
		exit(-1);
	}

	for (i = 0; i<monitor->nr_files; i++){
		char ***location = hdfsGetHosts(fs, monitor->file[i], 0, 64000000);;
		int k;
		for (k = 0; location[0][k]; k++){
			int host_id = DCscheduler_getHostID(monitor, location[0][k]);
			int file_id = i;
			monitor->map[file_id][k] = host_id;
			if (host_id == -1)
				printf(" location[0][k]= %s", location[0][k]);
		}
	}
	return monitor;
}



/*
struct DCmonitor *DCmonitor_create(char *host[], char *file, int nfiles, int rep)
{
	assert(host);
	assert(file);

	struct DCmonitor *monitor=(struct DCmonitor *)malloc(sizeof(*monitor));
	assert(monitor);
	memset(monitor, 0, sizeof(*monitor));

	int i,len;
	for(i=0;host[i];i++);
	monitor->nr_hosts=i;
	monitor->nr_files=nfiles;
	monitor->rep=rep;

	monitor->host = (char **)malloc(sizeof(char *) * (monitor->nr_hosts+1));
	assert(monitor->host);
	memset(monitor->host, 0,sizeof(char *) * (monitor->nr_hosts+1));
	for(i=0; i<monitor->nr_hosts; i++){
		len = strlen(host[i]);
		monitor->host[i] = (char *)malloc(sizeof(char) * (len+1));
		assert(monitor->host[i]);
		memset(monitor->host[i], 0, sizeof(char) * (len+1));
		strcpy(monitor->host[i], host[i]);
	}
	
	monitor->file = (char **)malloc(sizeof(char *) * (monitor->nr_files+1));
	assert(monitor->file);
	memset(monitor->file, 0, sizeof(char *) * (monitor->nr_files+1));
	for(i=0; i<monitor->nr_files; i++){

		char fileName[80];
		len = sprintf(fileName, "%s%d\n", file, i);

		monitor->file[i] = (char *)malloc(sizeof(char) * (len+1));
		assert(monitor->file[i]);
		memset(monitor->file[i], 0, sizeof(char) * (len+1));
		strcpy(monitor->file[i], fileName);
	}

	monitor->map = (int **)malloc(sizeof(int *) * (monitor->nr_files+1));
	assert(monitor->map);
	memset(monitor->map, 0, sizeof(int *) * (monitor->nr_files+1));
	for(i=0; i<monitor->nr_files; i++){
		monitor->map[i] = (int *)malloc(sizeof(int) * (monitor->rep));
		assert(monitor->map[i]);
		memset(monitor->map[i], 0, sizeof(int) * (monitor->rep));
	}

	hdfsFS fs = hdfsConnect("default", 0);
   	if(!fs) {
       	fprintf(stderr, "Oops! Failed to connect to hdfs!\n");
       	exit(-1);
   	} 

 	for(i=0; i<monitor->nr_files; i++){
		char ***location = hdfsGetHosts(fs, monitor->file[i], 0, 64000000);;
		int k;
		for(k=0;location[0][k];k++){
			int host_id = DCscheduler_getHostID(monitor, location[0][k]);
			int file_id = i;
			monitor->map[file_id][k] = host_id;
		}
	}	
	return monitor;
}
*/

void DCscheduler_showmap(struct DCmonitor *obj)
{
	assert(obj);
	int h,f;
	printf("********SHOW MAP ***********\n");
	for(f=0; f<obj->nr_files; f++){
	//	printf("file %s:\t", obj->file[f]);
		for(h=0; h<obj->rep; h++){
			printf("\t%d", obj->map[f][h]);
		}
		printf("\n");
	}
}

void DCmonitor_destroy(struct DCmonitor *obj)
{
	/* free map */
	int i;
	for(i=0; i<obj->nr_files; i++){
		free(obj->map[i]);
	}
	free(obj->map);

	/* free file */
	for(i=0; i<obj->nr_files; i++){
		free(obj->file[i]);
	}
	free(obj->file);

	/* free host */
	for(i=0; i<obj->nr_hosts; i++){
		free(obj->host[i]);
	}
	free(obj->host);

	/* free obj */
	free(obj);
}

char *HOSTS[]={
"marmot67.marmot.pdl.cmu.local",
"marmot128.marmot.pdl.cmu.local",
"marmot118.marmot.pdl.cmu.local",
"marmot112.marmot.pdl.cmu.local",
"marmot90.marmot.pdl.cmu.local",
"marmot23.marmot.pdl.cmu.local",
"marmot74.marmot.pdl.cmu.local",
"marmot87.marmot.pdl.cmu.local",
"marmot68.marmot.pdl.cmu.local",
"marmot129.marmot.pdl.cmu.local",
"marmot57.marmot.pdl.cmu.local",
"marmot70.marmot.pdl.cmu.local",
"marmot48.marmot.pdl.cmu.local",
"marmot4.marmot.pdl.cmu.local",
"marmot39.marmot.pdl.cmu.local",
"marmot46.marmot.pdl.cmu.local",
"marmot76.marmot.pdl.cmu.local",
"marmot7.marmot.pdl.cmu.local",
"marmot47.marmot.pdl.cmu.local",
"marmot79.marmot.pdl.cmu.local",
"marmot98.marmot.pdl.cmu.local",
"marmot16.marmot.pdl.cmu.local",
"marmot96.marmot.pdl.cmu.local",
"marmot110.marmot.pdl.cmu.local",
"marmot28.marmot.pdl.cmu.local",
"marmot63.marmot.pdl.cmu.local",
"marmot88.marmot.pdl.cmu.local",
"marmot14.marmot.pdl.cmu.local",
"marmot12.marmot.pdl.cmu.local",
"marmot65.marmot.pdl.cmu.local",
"marmot77.marmot.pdl.cmu.local",
"marmot49.marmot.pdl.cmu.local",
"marmot93.marmot.pdl.cmu.local",
"marmot73.marmot.pdl.cmu.local",
"marmot21.marmot.pdl.cmu.local",
"marmot40.marmot.pdl.cmu.local",
"marmot50.marmot.pdl.cmu.local",
"marmot32.marmot.pdl.cmu.local",
"marmot20.marmot.pdl.cmu.local",
"marmot9.marmot.pdl.cmu.local",
"marmot72.marmot.pdl.cmu.local",
"marmot91.marmot.pdl.cmu.local",
"marmot115.marmot.pdl.cmu.local",
"marmot52.marmot.pdl.cmu.local",
"marmot126.marmot.pdl.cmu.local",
"marmot29.marmot.pdl.cmu.local",
"marmot117.marmot.pdl.cmu.local",
"marmot84.marmot.pdl.cmu.local",
"marmot19.marmot.pdl.cmu.local",
"marmot38.marmot.pdl.cmu.local",
"marmot60.marmot.pdl.cmu.local",
"marmot100.marmot.pdl.cmu.local",
"marmot30.marmot.pdl.cmu.local",
"marmot42.marmot.pdl.cmu.local",
"marmot80.marmot.pdl.cmu.local",
"marmot51.marmot.pdl.cmu.local",
"marmot6.marmot.pdl.cmu.local",
"marmot78.marmot.pdl.cmu.local",
"marmot99.marmot.pdl.cmu.local",
"marmot122.marmot.pdl.cmu.local",
"marmot35.marmot.pdl.cmu.local",
"marmot83.marmot.pdl.cmu.local",
"marmot107.marmot.pdl.cmu.local",
"marmot64.marmot.pdl.cmu.local",
	NULL
};

/*
int main(int argc, char *argv[])
{
	printf("Hello World!\n");
	
	char* FileName = argv[1];
	int nFiles = strtoul(argv[2], NULL, 10);
	int nCopies = strtoul(argv[3], NULL, 10);

	struct DCmonitor *monitor = DCmonitor_create(HOSTS,FileName,nFiles,nCopies);
	DCscheduler_showmap(monitor);

	DCmonitor_destroy(monitor);
	return 0;
}
*/
